package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2SourcceApi {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("api")
      .getOrCreate()

    //导入隐式转换
    import spark.implicits._
    //导入所有的函数
    import org.apache.spark.sql.functions._


    /**
      * 1、读取csv格式的文件
      *
      */

    val studentsDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",") //默认分割方式是逗号
      .schema("id STRING, name STRING , age INT ,gender STRING ,clazz STRING")
      .load("data/students.txt")

    //studentsDF.show()

    /**
      * 2、读取一个json格式的文件
      * json 自带列名和类型，spark 会自动解析，不需要我们手动指定
      */

    val jsonDF: DataFrame = spark
      .read
      .format("json")
      .load("data/students.json")

    //打印表结构
    //jsonDF.printSchema()

    //jsonDF.show()

    /**
      * 3、使用jdbc 构建df
      *
      * spark 会继承数据库中的表结构，不需要手动指定
      */


    val jdbcDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable", "bigdata.student")
      .option("user", "root")
      .option("password", "123456")
      .load()

    //jdbcDF.printSchema()
    //jdbcDF.show()


    /**
      * parquet 格式的文件， 是一种压缩的格式，同时会自带列名，可以和hive完全兼容
      *
      */

    //生成一个parquet格式的文件
    /*studentsDF
      .write
      .mode(SaveMode.Overwrite)
      .format("parquet") //指定保存格式为parquet
      .save("data/parquet")*/

    /**
      * 4、读取parquet格式的文件
      *
      */
    val parquetDF: DataFrame = spark
      .read
      .format("parquet")
      .load("data/parquet")

    parquetDF.printSchema()
    parquetDF.show()


    /**
      *
      * 5、保存数据
      * json 格式和 parquet 格式不需要指定分割方式
      *
      */
    parquetDF
      .write
      .mode(SaveMode.Overwrite) //指定覆盖数据
      .format("csv")
      .option("sep", "\t") //csv  格式的分割方式模式是逗号，可以自定义
      .save("data/csv")



  }

}
